package streaming

import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object WindowDemo extends App {
  val sparkConf = new SparkConf()
  sparkConf.setAppName("Window")
  sparkConf.setMaster("local[4]")
  val sparkContext = new SparkContext(sparkConf)
  /**
   * 5s 一个窗口 即一个批次
   */
  val ssc = new StreamingContext(sparkContext, Durations.seconds(5))
  ssc.checkpoint("data/WindowDemo")
  val ds = ssc.socketTextStream("node1", 8888, StorageLevel.MEMORY_ONLY)

  val fun = (x: Int, y: Int) => x + y

  val mapDS = ds.flatMap(_.split(",")).map((_, 1))
  val countDS = mapDS.reduceByKeyAndWindow(fun, Durations.seconds(15), Durations.seconds(5))
  countDS.print()
  ssc.start()
  ssc.awaitTermination()
  ssc.stop()

}
